package cn.doitedu.rtdw.feature_etl;

import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @Author: 深似海
 * @Site: <a href="www.51doit.com">多易教育</a>
 * @QQ: 657270652
 * @Date: 2023/12/19
 * @Desc: 学大数据，上多易教育
 *   广告点击率预估特征数据加工
 **/
public class AdClickRationPredictFeatures {
    public static void main(String[] args) {


        // 创建编程环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointStorage("file:///d:/ckpt");
        env.setParallelism(1);
        env.setRuntimeMode(RuntimeExecutionMode.STREAMING);

        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);


        // 创建kafka中dwd行为明细topic的映射表
        tenv.executeSql(
                " create table dwd_events_kafka (                                 "+
                        "     event_id     string                                 "+
                        "     ,action_time   bigint                               "+
                        "     ,properties  map<string,string>                     "+
                        "     ,user_id  bigint                                    "+
                        "     ,rt as to_timestamp_ltz(action_time,3)              "+
                        "     ,watermark for rt as rt - interval '0' second       "+
                        " ) with (                                                "+
                        "     'connector' = 'kafka',                              "+
                        "     'topic' = 'dwd-events',                             "+
                        "     'properties.bootstrap.servers' = 'doitedu:9092',    " +
                        "     'properties.group.id' = 'doit43-2',                 " +
                        "     'scan.startup.mode' = 'latest-offset',              " +
                        "     'value.format'='json',                              " +
                        "     'value.json.fail-on-missing-field'='false',         " +
                        "     'value.fields-include' = 'EXCEPT_KEY'   )           "
        );

        // 利用flink-cep在行为日志流中寻找“曝光”后面有“点击”的模式
        tenv.executeSql(
                " with ad_events AS (                                       \n "+
                        "     SELECT                                                \n "+
                        "         user_id,                                          \n "+
                        " 		event_id,                                           \n "+
                        " 		action_time,                                        \n "+
                        " 		properties['ad_id'] as ad_id,                       \n "+
                        " 		properties['ad_tracking_id'] as ad_tracking_id,     \n "+
                        " 		rt                                                  \n "+
                        "     from dwd_events_kafka                                 \n "+
                        "     where event_id in ('ad_show','ad_click')              \n "+
                        " )                                                         \n "+
                        "                                                           \n "+
                        "                                                           \n "+
                        " SELECT                                                    \n "+
                        "      *                                                    \n "+
                        " from ad_events                                            \n "+
                        " match_recognize(                                          \n "+
                        "     partition by user_id,ad_tracking_id                   \n "+
                        " 	order by rt                                             \n "+
                        " 	measures                                                \n "+
                        " 	    A.user_id as uid,                                   \n "+
                        " 		A.ad_id as adid,                                    \n "+
                        " 		A.ad_tracking_id as tkid,                           \n "+
                        " 		A.event_id as show_event,                           \n "+
                        " 		A.action_time as show_time,                         \n "+
                        " 		B.event_id as click_event,                          \n "+
                        " 		B.action_time as click_time                         \n "+
                        " 	one row per match                                       \n "+
                        " 	after match skip past last row                          \n "+
                        " 	pattern (A B)                                           \n "+
                        " 	define                                                  \n "+
                        " 	   A AS A.event_id = 'ad_show',                         \n "+
                        " 	   B AS B.event_id = 'ad_click'                         \n "+
                        " )                                                         \n "
        ).print();
    }


}
